package ex.datastream;

import ex.datastream.functions.richFunction.KeyedStateReaderFunc;
import ex.vo.KeyedState;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.state.api.OperatorIdentifier;
import org.apache.flink.state.api.SavepointReader;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


public class ReadStateTest01 {

	public static void main(String[] args) throws Exception {
		//创建执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


		SavepointReader savepointer = SavepointReader.read(env, "e://ck/7166b3e58c29b28f2afd55bf601a0c00/chk-8");


		DataStream<KeyedState> keyedState = savepointer.readKeyedState(OperatorIdentifier.forUid("my-uid"), new KeyedStateReaderFunc());

		keyedState.flatMap(new FlatMapFunction<KeyedState, Object>() {

			@Override
			public void flatMap(KeyedState keyedState, Collector<Object> collector) throws Exception {
				collector.collect(String.format("key:{%s},value:{%s},times:{%s}",keyedState.key,keyedState.value,keyedState.times));
			}
		}).print();

		env.execute();

	}



}
